package com.ky.pro.big.data.hbase;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.NamespaceDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.util.ArrayList;
import java.util.List;

public class KyHBaseMain {

    public static void main(String[] args) throws Exception {
        // 建立连接
        Configuration configuration = HBaseConfiguration.create();
        configuration.set("hbase.zookeeper.quorum", "localhost");
        configuration.set("hbase.zookeeper.property.clientPort", "2181");
        configuration.set("hbase.rpc.timeout", "1000");
        configuration.set("hbase.client.retries.number", "1");
        configuration.set("hbase.client.pause", "100");
        Connection conn = ConnectionFactory.createConnection(configuration);
        Admin admin = conn.getAdmin();

        TableName tableName = TableName.valueOf("kevin_yang", "student");

        boolean isNameSpaceExist = false;
        for (NamespaceDescriptor namespaceDescriptor : admin.listNamespaceDescriptors()) {
            if ("kevin_yang".equals(namespaceDescriptor.getName())) {
                isNameSpaceExist = true;
                break;
            }
        }

        // 创建namespace
        if (isNameSpaceExist) {
            System.out.println("Namespace already exists");
        } else {
            admin.createNamespace(NamespaceDescriptor.create("kevin_yang").build());
        }

        // 建表
        if (admin.tableExists(tableName)) {
            System.out.println("Table already exists");
        } else {

            TableDescriptorBuilder tableDescriptorBuilder = TableDescriptorBuilder.newBuilder(tableName);

            List<ColumnFamilyDescriptor> cfDescriptorList = new ArrayList<>();
            cfDescriptorList.add(ColumnFamilyDescriptorBuilder.newBuilder("info".getBytes()).build());
            cfDescriptorList.add(ColumnFamilyDescriptorBuilder.newBuilder("score".getBytes()).build());

            tableDescriptorBuilder.setColumnFamilies(cfDescriptorList);

            admin.createTable(tableDescriptorBuilder.build());
            System.out.println("Table create successful");
        }

        String rowKey = "kevin_yang";

        // 插入数据
        Put put = new Put(Bytes.toBytes(rowKey)); // row key
        put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("student_id"), Bytes.toBytes("G20210698010089"));
        put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("class"), Bytes.toBytes("Pro Big Data"));
        put.addColumn(Bytes.toBytes("score"), Bytes.toBytes("understanding"), Bytes.toBytes("85"));
        put.addColumn(Bytes.toBytes("score"), Bytes.toBytes("programming"), Bytes.toBytes("80"));
        conn.getTable(tableName).put(put);
        System.out.println("Data insert success");

        // 查看数据
        Get get = new Get(Bytes.toBytes(rowKey));
        if (!get.isCheckExistenceOnly()) {
            Result result = conn.getTable(tableName).get(get);
            System.out.println("Getting Data Using Row Key:" + rowKey);
            for (Cell cell : result.rawCells()) {
                String colName = Bytes.toString(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength());
                String value = Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
                System.out.println("Data get success, colName: " + colName + ", value: " + value);
            }
        }

        // 删除数据
        Delete delete = new Delete(Bytes.toBytes(rowKey));      // 指定rowKey
        conn.getTable(tableName).delete(delete);
        System.out.println("Delete Success");

        // 删除表
        if (admin.tableExists(tableName)) {
            admin.disableTable(tableName);
            admin.deleteTable(tableName);
            System.out.println("Table Delete Successful");
        } else {
            System.out.println("Table does not exist!");
        }

        conn.close();

    }

}
